package com.aiobc.mq_transaction;


import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

/**
 * 事务消息的生产者跟我们之前的普通生产者的不同：
 *
 * a、需创建事务类型的生产者TransactionMQProducer；
 * b、需调用setTransactionListener()方法设置事务监听器；
 * c、使用sendMessageInTransaction()以事务方式发送消息；
 */
@Component
public class TransactionProducer {
    private static final String topic = "xc_topic";
    public void sendTransactionMsg(String msgBody) throws MQClientException, InterruptedException {
        // 创建事务类型的生产者
        TransactionMQProducer producer = new TransactionMQProducer("xc_producer");
        // 设置NameServer的地址
        producer.setNamesrvAddr("123.60.111.182:9876");
        // 设置事务监听器
        producer.setTransactionListener(new TransactionListenerImpl());
        //设置消息发送失败的最大重试次数
        producer.setRetryTimesWhenSendFailed(3);
        // 启动生产者
        producer.start();
        // 发送10条消息
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = new Message(topic, "", (msgBody + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 设置消息Key
                msg.setKeys("Num" + i);
                // 使用事务方式发送消息
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.println("sendResult = " + sendResult);
                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }

        // 阻塞，目的是为了在消息发送完成后才关闭生产者
        Thread.sleep(10000);
        producer.shutdown();
    }
}
